from flask import Flask, request, jsonify
from flask_cors import CORS
import os
import json
import requests
from typing import Any, Dict, List, Optional, Callable, Tuple
from abc import ABC, abstractmethod
from openai import OpenAI
import traceback
import ast
import re
import uuid
from werkzeug.utils import secure_filename
import shutil
import dashscope
import chromadb
from http import HTTPStatus

# 导入OBS Python SDK中的ObsClient类
from obs import ObsClient

app = Flask(__name__)

# 配置CORS以允许来自任何源的请求。
CORS(app, resources={r"/*": {"origins": "*"}})

# 用于存储工作流、API密钥和知识库的全局字典/文件
workflow_store = {}
KNOWLEDGE_BASE_DIR = "knowledge_bases"
KB_STORE_FILE = "knowledge_base_store.json"
knowledge_base_store = {}

def load_kb_store():
    """在服务器启动时加载知识库元数据"""
    global knowledge_base_store
    if os.path.exists(KB_STORE_FILE):
        try:
            with open(KB_STORE_FILE, 'r', encoding='utf-8') as f:
                knowledge_base_store = json.load(f)
        except (json.JSONDecodeError, IOError) as e:
            print(f"Warning: Could not load knowledge base store: {e}")
            knowledge_base_store = {}
    os.makedirs(KNOWLEDGE_BASE_DIR, exist_ok=True)

def save_kb_store():
    """将知识库元数据保存到文件"""
    with open(KB_STORE_FILE, 'w', encoding='utf-8') as f:
        json.dump(knowledge_base_store, f, indent=2, ensure_ascii=False)

class KnowledgeBaseHandler:
    """处理单个知识库的创建、索引、检索和删除"""
    def __init__(self, collection_name: str, persist_path: str):
        self.api_key = os.getenv("DASHSCOPE_API_KEY")
        if not self.api_key:
            self.api_key = 'sk-ad76a80e1feb400b9b7e505b597b3892'
        dashscope.api_key = self.api_key
        
        self.persist_path = persist_path
        self.collection_name = collection_name
        self.client = chromadb.PersistentClient(path=self.persist_path)
        self.collection = self.client.get_or_create_collection(
            name=self.collection_name,
            metadata={"hnsw:space": "cosine"}
        )

    def _embed_text(self, texts: List[str]) -> List[List[float]]:
        valid_texts = [t for t in texts if t.strip()]
        if not valid_texts:
            return []
        
        resp = dashscope.MultiModalEmbedding.call(
            model="multimodal-embedding-v1",
            input=[{'text': t} for t in valid_texts]
        )
        if resp.status_code == HTTPStatus.OK:
            return [item['embedding'] for item in resp.output['embeddings']]
        else:
            print(f"Embedding API调用失败: {resp.code} - {resp.message}")
            raise RuntimeError(f"无法获取文本向量: {resp.message}")

    def index_documents(self, documents: List[str]):
        if not documents:
            return
        embeddings = self._embed_text(documents)
        if not embeddings:
            return
        ids = [f"doc_{uuid.uuid4().hex}" for _ in range(len(documents))]
        self.collection.add(embeddings=embeddings, documents=documents, ids=ids)

    def retrieve(self, query: str, top_k: int = 10, score_threshold: float = 0.0) -> List[Dict[str, Any]]:
        query_embedding = self._embed_text([query])
        
        results = self.collection.query(
            query_embeddings=query_embedding,
            n_results=top_k,
            include=['documents', 'distances']
        )
        
        retrieved_docs = results['documents'][0]
        distances = results['distances'][0]
        
        similarities = [1 - d for d in distances]
        
        final_results = [
            {"text": doc, "score": sim} 
            for doc, sim in zip(retrieved_docs, similarities) 
            if sim >= score_threshold
        ]
        
        return final_results

    def get_all_documents(self) -> List[str]:
        results = self.collection.get(include=['documents'])
        return results.get('documents', [])

    def delete(self):
        try:
            self.client.delete_collection(name=self.collection_name)
        except ValueError as e:
            print(f"Warning: Could not delete collection '{self.collection_name}': {e}")
        
        if os.path.exists(self.persist_path):
            shutil.rmtree(self.persist_path)
            print(f"Deleted knowledge base directory: {self.persist_path}")

def upload_to_obs_and_get_url(
    access_key,
    secret_key,
    server_endpoint,
    bucket_name,
    local_file_path,
    object_key
):
    obs_client = None
    try:
        obs_client = ObsClient(
            access_key_id=access_key,
            secret_access_key=secret_key,
            server=server_endpoint
        )
        print(f"开始上传文件: {local_file_path} 到桶: {bucket_name} as {object_key}")
        resp = obs_client.putFile(
            bucketName=bucket_name,
            objectKey=object_key,
            file_path=local_file_path
        )
        if resp.status < 300:
            print("上传成功！")
            file_url = f"https://{bucket_name}.{server_endpoint}/{object_key}"
            return file_url
        else:
            print("上传失败。")
            print('错误码:', resp.errorCode)
            print('错误信息:', resp.errorMessage)
            return None
    except Exception as e:
        print(f"发生异常: {e}")
        traceback.print_exc()
        return None
    finally:
        if obs_client:
            obs_client.close()

@app.route('/upload-file', methods=['POST'])
def upload_file():
    ak = "HPUAUN3BNYFGS7SUUQBE"
    sk = "3RELn6SpPOtS9bXKwEfyz4jYBd4L1ghlvggWeUaz"
    endpoint = "obs.cn-north-4.myhuaweicloud.com"
    bucket = "obs-xc-bj"

    if not all([ak, sk, endpoint, bucket]):
        return jsonify({"error": "OBS configuration is missing on the server (OBS_AK, OBS_SK, OBS_ENDPOINT, OBS_BUCKET)."}), 500

    if 'file' not in request.files:
        return jsonify({"error": "No file part in the request"}), 400
    
    file = request.files['file']
    if file.filename == '':
        return jsonify({"error": "No file selected"}), 400

    if file:
        filename = secure_filename(file.filename)
        ext = os.path.splitext(filename)[1]
        object_key = f"uploads/{uuid.uuid4()}{ext}"
        
        temp_dir = "temp_uploads"
        os.makedirs(temp_dir, exist_ok=True)
        temp_path = os.path.join(temp_dir, filename)
        file.save(temp_path)

        try:
            uploaded_url = upload_to_obs_and_get_url(
                access_key=ak,
                secret_key=sk,
                server_endpoint=endpoint,
                bucket_name=bucket,
                local_file_path=temp_path,
                object_key=object_key
            )
        finally:
            if os.path.exists(temp_path):
                os.remove(temp_path)

        if uploaded_url:
            return jsonify({"url": uploaded_url}), 200
        else:
            return jsonify({"error": "Failed to upload to OBS"}), 500

    return jsonify({"error": "An unexpected error occurred"}), 500

class WorkflowNode(ABC):
    def __init__(self, node_id, name="", output_var="output"):
        self.node_id = node_id
        self.name = name or node_id
        self.output_var = output_var
        self.inputs = {}
        self.outputs = {}
        self.next_nodes = []
        
    @abstractmethod
    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        pass
    
    def add_next_node(self, node: 'WorkflowNode', condition: Optional[Callable] = None):
        self.next_nodes.append({'node': node, 'condition': condition})
    
    def set_input(self, key: str, value: Any):
        self.inputs[key] = value
    
    def get_output(self, key: str) -> Any:
        return self.outputs.get(key)

def replace_vars(text: str, context: Dict[str, Any]) -> str:
    if not text or not isinstance(text, str):
        return text
    def replace_match(match):
        var_name = match.group(1)
        for key, value in context.items():
            if isinstance(value, dict):
                if var_name in value and isinstance(value[var_name], dict) and 'value' in value[var_name]:
                    return str(value[var_name]['value'])
                elif var_name in value:
                    return str(value[var_name])
            if key == var_name:
                return str(value['value'] if isinstance(value, dict) and 'value' in value else value)
        return match.group(0)
    return re.sub(r'{{(\w+)}}', replace_match, text)

class StartNode(WorkflowNode):
    def __init__(self, node_id, name="开始", variables=None, output_var="output"):
        super().__init__(node_id, name, output_var)
        self.variables = variables or {}
    
    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        try:
            start_output = {}
            for key, var in self.variables.items():
                var_value = replace_vars(var['value'], context)
                start_output[key] = {"value": var_value, "type": var.get('type', 'text')}
            context[self.output_var] = start_output
            print(f"[{self.name}] 变量: {start_output}")
            return context
        except Exception as e:
            print(f"[{self.name}] 执行出错: {str(e)}")
            self.outputs['error'] = str(e)
            return context

class LLMNode(WorkflowNode):
    def __init__(self, node_id, name="LLM", model_url="", model_name="qwen-vl-max-latest", api_key=None, system="", user="", video_url="", output_var="output"):
        super().__init__(node_id, name, output_var)
        self.model_url = model_url
        self.model_name = model_name
        self.api_key = api_key
        self.system = system
        self.user = user
        self.video_url = video_url
    
    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        try:
            print(f"[{self.name}] 上下文: {json.dumps(context, ensure_ascii=False, indent=2)}")
            model_url = replace_vars(self.model_url, context)
            model_name = replace_vars(self.model_name, context)
            api_key = replace_vars(self.api_key, context) or os.environ.get("DASHSCOPE_API_KEY")
            if not api_key:
                print("警告: DASHSCOPE_API_KEY 未设置或未提供 API Key，请在环境变量或节点配置中设置。")
            system_prompt = replace_vars(self.system, context) if self.system else "You are a helpful assistant."
            user_prompt = replace_vars(self.user, context) if self.user else ''
            video_url = replace_vars(self.video_url, context)
            self.client = OpenAI(
                api_key=api_key,
                base_url=model_url or "https://dashscope.aliyuncs.com/compatible-mode/v1",
            )
            media_url = None
            media_type = None
            SUPPORTED_TYPES = ['text', 'document', 'image', 'video']
            if video_url:
                print(f"[{self.name}] 尝试获取变量 {video_url}")
                for key, value in context.items():
                    if isinstance(value, dict) and video_url in value:
                        media_url = value[video_url].get('value')
                        media_type = value[video_url].get('type')
                        print(f"[{self.name}] 从嵌套结构中找到 {video_url}: value={media_url}, type={media_type}")
                        break
                    if key == video_url:
                        media_url = value['value'] if isinstance(value, dict) and 'value' in value else value
                        media_type = value['type'] if isinstance(value, dict) and 'type' in value else None
                        print(f"[{self.name}] 从顶层找到 {video_url}: value={media_url}, type={media_type}")
                        break
                if media_url is None:
                    print(f"[{self.name}] 警告: 未输入变量 {video_url}")
                    raise ValueError(f"必需的变量 {video_url} 未在上下文中找到")
                if media_type is None:
                    print(f"[{self.name}] 警告: 变量 {video_url} 的类型未定义")
                    raise ValueError(f"变量 {video_url} 的类型未定义")
                if media_type not in SUPPORTED_TYPES:
                    print(f"[{self.name}] 警告: 变量 {video_url} 的类型 {media_type} 不在支持的类型列表 {SUPPORTED_TYPES} 中")
                    raise ValueError(f"不支持的媒体类型: {media_type}")
            print(f"[{self.name}] 替换后的提示词: system={system_prompt}, user={user_prompt}")
            messages = [
                {"role": "system", "content": [{"type": "text", "text": system_prompt}]}
            ]
            user_content = []
            if video_url and media_url:
                content_type = "text"
                if media_type == 'image':
                    content_type = "image_url"
                elif media_type == 'video':
                    content_type = "video_url"
                elif media_type == 'document':
                    content_type = "text"
                user_content.append({
                    "type": content_type,
                    content_type: {"url": media_url} if content_type in ['image_url', 'video_url'] else {"text": media_url}
                })
                print(f"[{self.name}] 添加媒体字段: type={content_type}, value={media_url}")
            if user_prompt:
                user_content.append({"type": "text", "text": user_prompt})
            if user_content:
                messages.append({"role": "user", "content": user_content})
            print(f"[{self.name}] 调用模型: model={model_name}, messages={json.dumps(messages, ensure_ascii=False, indent=2)}")
            completion = self.client.chat.completions.create(
                model=model_name,
                messages=messages
            )
            result = completion.choices[0].message.content
            llm_output = {
                'response': result,
                'raw_completion': str(completion)
            }
            context[self.output_var] = llm_output
            print(f"[{self.name}] 模型分析完成，结果存储在: {self.output_var}, 响应: {result}")
            return context
        except Exception as e:
            print(f"[{self.name}] 执行出错: {str(e)}")
            traceback.print_exc()
            self.outputs['error'] = str(e)
            return context

class BranchNode(WorkflowNode):
    def __init__(self, node_id, name="条件分支", variable="", condition_type="", condition_value="", output_var="output"):
        super().__init__(node_id, name, output_var)
        self.variable = variable
        self.condition_type = condition_type
        self.condition_value = condition_value
    
    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        try:
            variable = self.variable
            condition_type = self.condition_type
            condition_value = self.condition_value
            var_value = None
            for key, value in context.items():
                if isinstance(value, dict) and variable in value:
                    var_value = value[variable].get('value') if isinstance(value[variable], dict) else value[variable]
                    break
                if key == variable:
                    var_value = value['value'] if isinstance(value, dict) and 'value' in value else value
                    break
            var_value = str(var_value).strip() if var_value is not None else ''
            print(f"[{self.name}] 原始变量值: {var_value} (type: {type(var_value)})")
            condition_value = str(condition_value).strip()
            print(f"[{self.name}] 提取后的变量值: '{var_value}' (type: {type(var_value)}), 条件值: '{condition_value}' (type: {type(condition_value)})")
            if condition_type == 'contains':
                result = condition_value in var_value
            elif condition_type == 'not_contains':
                result = condition_value not in var_value
            elif condition_type == 'starts_with':
                result = var_value.startswith(condition_value)
            elif condition_type == 'ends_with':
                result = var_value.endswith(condition_value)
            elif condition_type == 'equals':
                result = var_value == condition_value
            elif condition_type == 'not_equals':
                result = var_value != condition_value
            elif condition_type == 'is_empty':
                result = not var_value
            elif condition_type == 'is_not_empty':
                result = bool(var_value)
            else:
                result = False
                print(f"[{self.name}] 未知的条件类型: {condition_type}")
            branch_output = {'condition_result': result}
            context[self.output_var] = branch_output
            print(f"[{self.name}] 分支判断结果: {result} (变量: {variable}, 条件: {condition_type}, 值: {condition_value})")
            return context
        except Exception as e:
            print(f"[{self.name}] 执行出错: {str(e)}")
            self.outputs['error'] = str(e)
            return context

class CodeNode(WorkflowNode):
    def __init__(self, node_id, code="", name="代码执行", output_var="output", input_variables=None):
        super().__init__(node_id, name, output_var)
        self.code = code
        self.input_variables = input_variables or {}

    def _find_main_function_name(self, code_str: str) -> Optional[str]:
        try:
            tree = ast.parse(code_str)
            for node in tree.body:
                if isinstance(node, ast.FunctionDef):
                    return node.name
        except SyntaxError:
            return None
        return None

    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        try:
            function_name = self._find_main_function_name(self.code)
            if not function_name:
                raise ValueError("未在提供的代码中找到有效的 Python 函数定义 (def a_function(...):)。")

            args = {}
            for local_name, upstream_var_name in self.input_variables.items():
                arg_value = None
                found = False
                if upstream_var_name in context:
                    arg_value = context[upstream_var_name]
                    found = True
                
                if not found:
                    for key, value in context.items():
                        if isinstance(value, dict) and upstream_var_name in value:
                            arg_value = value[upstream_var_name]
                            found = True
                            break
                
                if not found:
                    print(f"[{self.name}] 警告: 未在上下文中找到输入变量 '{upstream_var_name}' (本地参数名: '{local_name}')")
                    args[local_name] = None
                    continue

                if isinstance(arg_value, dict):
                    if 'response' in arg_value:
                        args[local_name] = arg_value['response']
                    elif 'json' in arg_value:
                        args[local_name] = arg_value['json']
                    elif 'value' in arg_value:
                        args[local_name] = arg_value['value']
                    else:
                        args[local_name] = arg_value
                else:
                    args[local_name] = arg_value

            print(f"[{self.name}] 准备传递给函数的参数: {list(args.keys())}")

            exec_globals = {'json': json, 'requests': requests}
            exec(self.code, exec_globals)

            user_func = exec_globals.get(function_name)
            if not callable(user_func):
                raise ValueError(f"在代码中未找到名为 '{function_name}' 的可调用函数。")

            result = user_func(**args)

            context[self.output_var] = result
            print(f"[{self.name}] 代码执行完成，返回值已存储在: {self.output_var}")
            return context
        except Exception as e:
            print(f"[{self.name}] 执行出错: {str(e)}")
            tb_str = traceback.format_exc()
            traceback.print_exc()
            self.outputs['error'] = f"{str(e)}\n\n{tb_str}"
            return context

class HttpNode(WorkflowNode):
    def __init__(self, node_id, method="GET", url="", headers="", data="", name="HTTP请求", output_var="output"):
        super().__init__(node_id, name, output_var)
        self.method = method.upper()
        self.url = url
        self.headers = headers
        self.data = data
    
    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        try:
            method = self.inputs.get('method', self.method)
            url = replace_vars(self.inputs.get('url', self.url), context)
            headers_input = self.inputs.get('headers', self.headers) or '{}'
            headers = json.loads(replace_vars(headers_input, context))
            data = replace_vars(self.inputs.get('data', self.data), context)
            if not url.startswith(('http://', 'https://')):
                url = 'http://' + url
                print(f"[{self.name}] 自动补充协议，URL更新为: {url}")
            headers = {k: replace_vars(v, context) for k, v in headers.items()}
            if data:
                try:
                    data = json.loads(data)
                except json.JSONDecodeError:
                    pass
            print(f"[{self.name}] 发送HTTP请求: {method} {url}")
            response = requests.request(
                method=method,
                url=url,
                headers=headers,
                json=data if method in ['POST', 'PUT'] and isinstance(data, dict) else None,
                params=data if method == 'GET' and not isinstance(data, dict) else None,
                timeout=30
            )
            response.raise_for_status()
            http_output = {
                'status_code': response.status_code,
                'json': response.json() if response.headers.get('Content-Type', '').startswith('application/json') else None,
                'text': response.text
            }
            context[self.output_var] = http_output
            print(f"[{self.name}] HTTP请求完成: {method} {url} - {response.status_code}")
            return context
        except Exception as e:
            print(f"[{self.name}] 执行出错: {str(e)}")
            self.outputs['error'] = str(e)
            return context

class KnowledgeNode(WorkflowNode):
    def __init__(self, node_id, name="知识检索", query_variable="", knowledge_base_id="", top_k=3, score_threshold=0.2, output_var="output"):
        super().__init__(node_id, name, output_var)
        self.query_variable = query_variable
        self.knowledge_base_id = knowledge_base_id
        self.top_k = top_k
        self.score_threshold = score_threshold

    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        try:
            if not self.knowledge_base_id:
                raise ValueError("未选择知识库")
            if not self.query_variable:
                raise ValueError("未配置查询变量")
            
            # 从上下文中获取查询文本
            query_text = None
            raw_query = None
            # 遍历上下文，查找包含 query_variable 的嵌套结构
            for key, value in context.items():
                if isinstance(value, dict) and self.query_variable in value:
                    raw_query = value[self.query_variable]
                    break
                if key == self.query_variable:
                    raw_query = value
                    break
            
            if raw_query is None:
                raise ValueError(f"查询变量 '{self.query_variable}' 在上下文中未找到")

            if isinstance(raw_query, dict):
                if 'response' in raw_query:
                    query_text = raw_query['response']
                elif 'value' in raw_query:
                    query_text = raw_query['value']
                elif 'text' in raw_query:
                    query_text = raw_query['text']
                else:
                    query_text = str(raw_query)
            else:
                query_text = str(raw_query)
            
            if not query_text:
                raise ValueError(f"查询变量 '{self.query_variable}' 在上下文中为空")

            print(f"[{self.name}] 使用查询: '{query_text}'")

            kb_info = knowledge_base_store.get(self.knowledge_base_id)
            if not kb_info:
                raise ValueError(f"ID为 '{self.knowledge_base_id}' 的知识库未找到")

            handler = KnowledgeBaseHandler(collection_name=kb_info['id'], persist_path=kb_info['path'])
            results = handler.retrieve(
                query_text, 
                top_k=int(self.top_k), 
                score_threshold=float(self.score_threshold)
            )
            
            context[self.output_var] = results
            print(f"[{self.name}] 检索到 {len(results)} 条结果 (TopK={self.top_k}, Score>={self.score_threshold})，已存入 '{self.output_var}'")
            return context
        except Exception as e:
            print(f"[{self.name}] 执行出错: {str(e)}")
            traceback.print_exc()
            self.outputs['error'] = str(e)
            return context

class EndNode(WorkflowNode):
    def __init__(self, node_id, name="结束", variables=None, output_var="output"):
        super().__init__(node_id, name, output_var)
        self.variables = variables or {}
    
    def execute(self, context: Dict[str, Any], all_edges: List[Dict] = None, all_nodes: Dict[str, 'WorkflowNode'] = None) -> Dict[str, Any]:
        try:
            resolved_variables = {}
            upstream_vars = self.get_upstream_variables(context, all_edges, all_nodes)
            
            for key in self.variables:
                if key in upstream_vars:
                    value = upstream_vars[key]
                    if isinstance(value, dict) and 'json' in value and 'status_code' in value:
                        if value['json'] is not None and isinstance(value['json'], dict):
                            resolved_variables.update(value['json'])
                        else:
                            resolved_variables[key] = 'HTTP 响应无 JSON 数据'
                    elif isinstance(value, dict) and 'response' in value:
                        resolved_variables['result'] = value['response']
                    elif isinstance(value, dict) and 'value' in value:
                        resolved_variables[key] = value['value']
                    else:
                        resolved_variables[key] = value
                else:
                    resolved_variables[key] = '未找到变量值'
            
            end_output = {
                'variables': resolved_variables
            }
            context[self.output_var] = end_output
            print(f"\n{'='*50}")
            print(f"[{self.name}] 分析完成!")
            print(f"输出变量: {resolved_variables}")
            print(f"{'='*50}\n")
            return context
        except Exception as e:
            print(f"[{self.name}] 执行出错: {str(e)}")
            self.outputs['error'] = str(e)
            return context
    
    def get_upstream_variables(self, context: Dict[str, Any], all_edges: List[Dict], all_nodes: Dict[str, 'WorkflowNode']) -> Dict[str, Any]:
        upstream_vars = {}
        visited = set()
        
        def collect_upstream_nodes(node_id):
            if node_id in visited:
                return
            visited.add(node_id)
            node = all_nodes.get(node_id)
            if node and node.output_var in context:
                upstream_vars[node.output_var] = context[node.output_var]
            
            incoming_edges = [edge for edge in all_edges if edge['target'] == node_id]
            for edge in incoming_edges:
                collect_upstream_nodes(edge['source'])
        
        collect_upstream_nodes(self.node_id)
        return upstream_vars

class WorkflowEngine:
    def __init__(self):
        self.nodes = {}
        self.all_edges = []
        self.node_type_counters = {
            'start': 0,
            'llm': 0,
            'branch': 0,
            'code': 0,
            'http': 0,
            'knowledge': 0,
            'end': 0
        }
    
    def add_node(self, node: WorkflowNode):
        self.nodes[node.node_id] = node
    
    def set_all_edges(self, edges_data: List[Dict]):
        self.all_edges = edges_data
    
    def topological_sort(self, nodes_data):
        graph = {node['id']: [] for node in nodes_data}
        in_degree = {node['id']: 0 for node in nodes_data}
        for edge in self.all_edges:
            graph[edge['source']].append((edge['target'], edge.get('sourceHandle')))
            in_degree[edge['target']] += 1
        queue = [node_id for node_id, degree in in_degree.items() if degree == 0]
        sorted_nodes = []
        while queue:
            current = queue.pop(0)
            sorted_nodes.append(current)
            for neighbor, _ in graph[current]:
                in_degree[neighbor] -= 1
                if in_degree[neighbor] == 0:
                    queue.append(neighbor)
        if len(sorted_nodes) != len(nodes_data):
            raise Exception("Graph has cycles")
        return sorted_nodes

    def execute_workflow(self, nodes_data, initial_context: Dict[str, Any] = None) -> Dict[str, Any]:
        context = initial_context or {}
        sorted_node_ids = self.topological_sort(nodes_data)
        executed_nodes = set()
        queue = [(sorted_node_ids[0], None)]

        while queue:
            node_id, source_handle = queue.pop(0)
            if node_id in executed_nodes:
                continue

            node = self.nodes[node_id]
            print(f"执行节点: {node.name} ({node_id})")
            executed_nodes.add(node_id)

            inputs = {}
            for edge in self.all_edges:
                if edge['target'] == node_id:
                    source_node = self.nodes.get(edge['source'])
                    if source_node:
                        source_output = context.get(source_node.output_var, {})
                        inputs[edge['source']] = source_output
            node.inputs = inputs
            
            if isinstance(node, EndNode):
                context = node.execute(context, self.all_edges, self.nodes)
            else:
                context = node.execute(context)
            
            context[f"{node_id}_outputs"] = node.outputs

            next_edges = [e for e in self.all_edges if e['source'] == node_id]
            if isinstance(node, BranchNode):
                condition_result = context.get(node.output_var, {}).get('condition_result', False)
                for edge in next_edges:
                    handle = edge.get('sourceHandle')
                    if (handle == 'true_branch' and condition_result) or (handle == 'false_branch' and not condition_result):
                        queue.append((edge['target'], handle))
            else:
                for edge in next_edges:
                    queue.append((edge['target'], edge.get('sourceHandle')))

        return context

NODE_TYPE_MAP = {
    "start": StartNode,
    "llm": LLMNode,
    "branch": BranchNode,
    "code": CodeNode,
    "http": HttpNode,
    "knowledge": KnowledgeNode,
    "end": EndNode,
}

@app.route('/knowledge-bases', methods=['GET'])
def get_knowledge_bases():
    return jsonify(list(knowledge_base_store.values()))

@app.route('/knowledge-bases/<kb_id>/documents', methods=['GET'])
def get_kb_documents(kb_id):
    try:
        if kb_id not in knowledge_base_store:
            return jsonify({"error": "Knowledge base not found"}), 404
        
        kb_info = knowledge_base_store[kb_id]
        handler = KnowledgeBaseHandler(collection_name=kb_info['id'], persist_path=kb_info['path'])
        
        documents = handler.get_all_documents()
        
        return jsonify({"documents": documents}), 200
    except Exception as e:
        traceback.print_exc()
        return jsonify({"error": str(e)}), 500

@app.route('/knowledge-bases', methods=['POST'])
def create_knowledge_base():
    try:
        name = request.form.get('name')
        source_type = request.form.get('sourceType')
        content = ""

        if not name or not source_type:
            return jsonify({"error": "Name and sourceType are required"}), 400

        if source_type == 'text':
            content = request.form.get('content', '')
        elif source_type == 'url':
            url = request.form.get('url', '')
            if not url:
                return jsonify({"error": "URL is required for sourceType 'url'"}), 400
            try:
                response = requests.get(url, timeout=20)
                response.raise_for_status()
                try:
                    content = response.content.decode('utf-8')
                except UnicodeDecodeError:
                    content = response.text
            except requests.RequestException as e:
                return jsonify({"error": f"Failed to fetch content from URL: {e}"}), 400
        elif source_type == 'file':
            if 'file' not in request.files:
                return jsonify({"error": "No file part in the request for sourceType 'file'"}), 400
            file = request.files['file']
            if file.filename == '':
                return jsonify({"error": "No file selected"}), 400
            try:
                content = file.read().decode('utf-8')
            except UnicodeDecodeError:
                 return jsonify({"error": "Failed to decode file. Please ensure it is UTF-8 encoded."}), 400
        else:
            return jsonify({"error": f"Invalid sourceType: {source_type}"}), 400
        
        if not content.strip():
            return jsonify({"error": "No content provided or fetched to index"}), 400

        docs = [doc.strip() for doc in content.split('\n') if doc.strip()]
        if not docs:
            return jsonify({"error": "No valid documents found in the provided content"}), 400

        kb_id = f"kb_{uuid.uuid4().hex}"
        persist_path = os.path.join(KNOWLEDGE_BASE_DIR, kb_id)
        os.makedirs(persist_path, exist_ok=True)
        
        handler = KnowledgeBaseHandler(collection_name=kb_id, persist_path=persist_path)
        handler.index_documents(docs)

        kb_info = {"id": kb_id, "name": name, "path": persist_path}
        knowledge_base_store[kb_id] = kb_info
        save_kb_store()
        
        return jsonify(kb_info), 201
    except Exception as e:
        traceback.print_exc()
        return jsonify({"error": str(e)}), 500

@app.route('/knowledge-bases/<kb_id>', methods=['DELETE'])
def delete_knowledge_base(kb_id):
    try:
        if kb_id not in knowledge_base_store:
            return jsonify({"error": "Knowledge base not found"}), 404
        
        kb_info = knowledge_base_store[kb_id]
        handler = KnowledgeBaseHandler(collection_name=kb_info['id'], persist_path=kb_info['path'])
        handler.delete()
        
        del knowledge_base_store[kb_id]
        save_kb_store()
        
        return jsonify({"message": "Knowledge base deleted successfully"}), 200
    except Exception as e:
        traceback.print_exc()
        return jsonify({"error": str(e)}), 500

@app.route('/save-workflow', methods=['POST'])
def save_workflow():
    try:
        data = request.get_json()
        if not data:
            return jsonify({"error": "No JSON data received"}), 400
        workflow_id = str(uuid.uuid4())
        api_key = str(uuid.uuid4())
        workflow_store[workflow_id] = {
            "config": data,
            "api_key": api_key
        }
        return jsonify({
            "workflow_id": workflow_id,
            "api_key": api_key,
            "endpoint": f"/api/workflows/{workflow_id}"
        }), 200
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route('/api/workflows/<workflow_id>', methods=['POST'])
def execute_workflow_api(workflow_id):
    if workflow_id not in workflow_store:
        return jsonify({"error": "Workflow not found"}), 404
    
    api_key = request.headers.get('X-API-Key')
    if not api_key or api_key != workflow_store[workflow_id]['api_key']:
        return jsonify({"error": "Invalid or missing API key"}), 403
    
    workflow_data = workflow_store[workflow_id]['config']
    initial_context = request.get_json() or {}
    
    try:
        engine = WorkflowEngine()
        nodes_data = workflow_data.get('nodes', [])
        edges_data = workflow_data.get('edges', [])
        
        for node_data in nodes_data:
            node_id = node_data.get('id')
            node_backend_type = node_data.get('data', {}).get('nodeType')
            node_name = node_data.get('data', {}).get('name', node_id)
            node_props = node_data.get('data', {})
            engine.node_type_counters[node_backend_type] += 1
            output_var = node_props.get('output_var', f"{node_backend_type}_node{engine.node_type_counters[node_backend_type]}_output")
            node_class = NODE_TYPE_MAP.get(node_backend_type)
            if node_class:
                node = None
                if node_class == LLMNode:
                    node = LLMNode(
                        node_id,
                        name=node_name,
                        model_url=node_props.get('model_url', ''),
                        model_name=node_props.get('model_name', 'qwen-plus'),
                        api_key=node_props.get('api_key', ''),
                        system=node_props.get('system', ''),
                        user=node_props.get('user', ''),
                        video_url=node_props.get('video_url', ''),
                        output_var=output_var
                    )
                elif node_class == BranchNode:
                    node = BranchNode(
                        node_id,
                        name=node_name,
                        variable=node_props.get('variable', ''),
                        condition_type=node_props.get('conditionType', ''),
                        condition_value=node_props.get('conditionValue', ''),
                        output_var=output_var
                    )
                elif node_class == CodeNode:
                    node = CodeNode(
                        node_id,
                        code=node_props.get('code', ''),
                        name=node_name,
                        output_var=output_var,
                        input_variables=node_props.get('inputVariables', {})
                    )
                elif node_class == HttpNode:
                    node = HttpNode(
                        node_id,
                        method=node_props.get('method', 'GET'),
                        url=node_props.get('url', ''),
                        headers=node_props.get('headers', '{}'),
                        data=node_props.get('data', ''),
                        name=node_name,
                        output_var=output_var
                    )
                elif node_class == KnowledgeNode:
                    node = KnowledgeNode(
                        node_id,
                        name=node_name,
                        query_variable=node_props.get('query_variable', ''),
                        knowledge_base_id=node_props.get('knowledge_base_id', ''),
                        top_k=node_props.get('top_k', 3),
                        score_threshold=node_props.get('score_threshold', 0.2),
                        output_var=output_var
                    )
                elif node_class == StartNode:
                    node = StartNode(node_id, name=node_name, variables=node_props.get('variables', {}), output_var=output_var)
                elif node_class == EndNode:
                    node = EndNode(node_id, name=node_name, variables=node_props.get('variables', {}), output_var=output_var)
                engine.add_node(node)
            else:
                return jsonify({"error": f"Unknown node type: {node_backend_type}"}), 400
        
        engine.set_all_edges(edges_data)
        final_context = engine.execute_workflow(nodes_data, initial_context)
        
        end_node_output_data = None
        for key, value in final_context.items():
            if key.startswith("end_") and key.endswith("_output"):
                end_node_output_data = value
                break

        if end_node_output_data and 'variables' in end_node_output_data:
            final_result = end_node_output_data.get('variables', {})
            return jsonify(final_result), 200
        else:
            return jsonify({
                "status": "error",
                "message": "Workflow executed, but no final output variables were found in the EndNode."
            }), 500

    except Exception as e:
        traceback.print_exc()
        return jsonify({"status": "error", "message": str(e), "traceback": traceback.format_exc()}), 500

@app.route('/execute-workflow', methods=['POST'])
def execute_workflow_local():
    try:
        data = request.get_json()
        if not data:
            return jsonify({"error": "No JSON data received"}), 400
        nodes_data = data.get('nodes', [])
        edges_data = data.get('edges', [])
        initial_context = data.get('initial_context', {})
        engine = WorkflowEngine()
        created_nodes = {}
        start_node_id = None
        for node_data in nodes_data:
            node_id = node_data.get('id')
            node_backend_type = node_data.get('data', {}).get('nodeType')
            node_name = node_data.get('data', {}).get('name', node_id)
            node_props = node_data.get('data', {})
            engine.node_type_counters[node_backend_type] += 1
            output_var = node_props.get('output_var', f"{node_backend_type}_node{engine.node_type_counters[node_backend_type]}_output")
            node_class = NODE_TYPE_MAP.get(node_backend_type)
            if node_class:
                node = None
                if node_class == LLMNode:
                    node = LLMNode(
                        node_id,
                        name=node_name,
                        model_url=node_props.get('model_url', ''),
                        model_name=node_props.get('model_name', 'qwen-plus'),
                        api_key=node_props.get('api_key', ''),
                        system=node_props.get('system', ''),
                        user=node_props.get('user', ''),
                        video_url=node_props.get('video_url', ''),
                        output_var=output_var
                    )
                elif node_class == BranchNode:
                    node = BranchNode(
                        node_id,
                        name=node_name,
                        variable=node_props.get('variable', ''),
                        condition_type=node_props.get('conditionType', ''),
                        condition_value=node_props.get('conditionValue', ''),
                        output_var=output_var
                    )
                elif node_class == CodeNode:
                    node = CodeNode(
                        node_id,
                        code=node_props.get('code', ''),
                        name=node_name,
                        output_var=output_var,
                        input_variables=node_props.get('inputVariables', {})
                    )
                elif node_class == HttpNode:
                    node = HttpNode(
                        node_id,
                        method=node_props.get('method', 'GET'),
                        url=node_props.get('url', ''),
                        headers=node_props.get('headers', '{}'),
                        data=node_props.get('data', ''),
                        name=node_name,
                        output_var=output_var
                    )
                elif node_class == KnowledgeNode:
                    node = KnowledgeNode(
                        node_id,
                        name=node_name,
                        query_variable=node_props.get('query_variable', ''),
                        knowledge_base_id=node_props.get('knowledge_base_id', ''),
                        top_k=node_props.get('top_k', 3),
                        score_threshold=node_props.get('score_threshold', 0.2),
                        output_var=output_var
                    )
                elif node_class == StartNode:
                    node = StartNode(node_id, name=node_name, variables=node_props.get('variables', {}), output_var=output_var)
                    start_node_id = node_id
                elif node_class == EndNode:
                    node = EndNode(node_id, name=node_name, variables=node_props.get('variables', {}), output_var=output_var)
                engine.add_node(node)
                created_nodes[node_id] = node
            else:
                print(f"错误: 未知的节点类型: {node_backend_type} (ID: {node_id})")
                return jsonify({"error": f"未知节点类型: {node_backend_type}"}), 400
        if not start_node_id:
            return jsonify({"error": "工作流定义中未找到 '开始节点' (StartNode)。一个工作流必须有一个 StartNode。"}), 400
        engine.set_all_edges(edges_data)
        final_context = engine.execute_workflow(nodes_data, initial_context)
        serializable_context = {}
        for key, value in final_context.items():
            try:
                if key.endswith('_outputs') and isinstance(value, dict) and 'raw_completion' in value:
                    temp_value = {k: v for k, v in value.items() if k != 'raw_completion'}
                    json.dumps(temp_value)
                    serializable_context[key] = temp_value
                else:
                    json.dumps(value)
                    serializable_context[key] = value
            except TypeError:
                serializable_context[key] = str(value)
        return jsonify({"status": "success", "context": serializable_context}), 200
    except Exception as e:
        traceback.print_exc()
        return jsonify({"status": "error", "message": str(e), "traceback": traceback.format_exc()}), 500

if __name__ == '__main__':
    load_kb_store()
    if not os.environ.get("DASHSCOPE_API_KEY"):
        print("\n" + "="*80)
        print("!!! 提示: DASHSCOPE_API_KEY 环境变量未设置，请设置你的专属密钥。 !!!")
        print("!!! 示例: export DASHSCOPE_API_KEY=\"sk-YOUR_ACTUAL_KEY_HERE\" !!!")
        print("="*80 + "\n")
    
    if not all([os.environ.get("OBS_AK"), os.environ.get("OBS_SK"), os.environ.get("OBS_ENDPOINT"), os.environ.get("OBS_BUCKET")]):
        print("\n" + "="*80)
        print("!!! 提示: OBS 配置环境变量未完全设置 (需要 OBS_AK, OBS_SK, OBS_ENDPOINT, OBS_BUCKET)。 !!!")
        print("!!! 文件上传功能将不可用，直到这些变量被设置。 !!!")
        print("!!! 示例: export OBS_AK=\"YOUR_AK\" !!!")
        print("="*80 + "\n")

    app.run(debug=True, host='0.0.0.0', port=5000)